// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tsd.service;

import com.google.common.base.Charsets;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import com.google.common.io.Files;
import com.stumbleupon.async.DeferredGroupException;
import net.opentsdb.core.IQuery;
import net.opentsdb.core.Internal;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.TSQuery;
import net.opentsdb.query.filter.TagVFilter;
import net.opentsdb.tsd.HttpQuery;
import net.opentsdb.tsd.http.ServiceResponse;
import net.opentsdb.uid.NoSuchUniqueName;
import net.opentsdb.utils.CharUtils;
import net.opentsdb.utils.Exceptions;
import org.hbase.async.KeyValue;
import org.hbase.async.Scanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.LongAdder;

/**
 *  导出数据请求
 */
public class ExportDataService extends RpcServiceImpl{

    private static final Logger LOG = LoggerFactory.getLogger(ExportDataService.class);

    // 导出最大值为10万
    private static final long LIMIT_MAX =  100000;

    private final TSQuery dataQuery;

    private LongAdder limit = new LongAdder();

    private long count = LIMIT_MAX;

    private File file;

    public ExportDataService(TSDB tsdb, HttpQuery query){
        super(tsdb, query);
        dataQuery = query.serializer().parseQueryV1();
        limit.add(LIMIT_MAX);
        file = null;
    }

    @Override
    public void doPost() {
        Map<String, String> res = Maps.newHashMap();
        InetAddress addr;

        try{
            addr = InetAddress.getLocalHost();

            long startTime = System.currentTimeMillis();
            Files.write("",file, Charsets.UTF_8);

            dataQuery.validateAndSetQuery();
            doDump();

            res.put("time",(System.currentTimeMillis()-startTime)+"ms");
            res.put("path",dataQuery.getPath());
            res.put("count",String.valueOf(count-limit.sum()));
            res.put("hostip",addr.getHostAddress());
            res.put("hostname",addr.getHostName());

            response = ServiceResponse
                    .builder(true)
                    .response(res)
                    .build();
        }catch (Exception e) {
            putErrorResponse(false, "export metric data error",e);
        }

    }

    @Override
    public void checks() throws Exception{
        if(Strings.isNullOrEmpty(dataQuery.getPath()))
            throw new IOException("请输入保存文件");

        file = new File(dataQuery.getPath());

        if(!file.exists()){
            LOG.debug(" {} file doesn't exists ,so create",dataQuery.getPath());

            if(!file.getParentFile().exists()) file.getParentFile().mkdirs();

            file.createNewFile();
        }
    }

    @Override
    public Object doParse(){

        // 判断导出限制，默认十万
        String limitParam = getQuery().getQueryStringParam("limit");

        if(CharUtils.isNumeric(limitParam)){
            long limitResult = Long.parseLong(limitParam);
            if(limitResult > 0 && limitResult< LIMIT_MAX){
                limit.reset();
                limit.add(limitResult);
                count = limitResult;
            }
        }

        return null;
    }


    /**
     * 导出数据
     *
     * @throws Exception
     */
    private void doDump() throws Exception{
        final StringBuilder buf = new StringBuilder();
        IQuery[] queries = dataQuery.buildQueriesAsync(tsdb).join();

        // 注意这里将所有的filters放在一起，如果一个请求包含多种请求将不适用
        List<TagVFilter> filters = new ArrayList<>();
        dataQuery.getQueries().forEach(query->filters.addAll(query.getFilters()));

        for (final IQuery query : queries) {
            final List<Scanner> scanners = Internal.getScanners(query);

            for (Scanner scanner : scanners) {
                ArrayList<ArrayList<KeyValue>> rows;

                while ((rows = scanner.nextRows().joinUninterruptibly()) != null) {

                    for (final ArrayList<KeyValue> row : rows) {

                        final byte[] key = row.get(0).key();
                        final long base_time = Internal.baseTime(tsdb, key);
                        final String metric = Internal.metricName(tsdb, key);
                        Map<String, String> tags = Internal.getTags(tsdb, key);

                        boolean flag = true;
                        if(filters.size() > 0){
                            // filters过滤
                            for (final TagVFilter filter : filters) {
                                if(!filter.match(tags).joinUninterruptibly()){
                                    flag = false;
                                    break;
                                }
                            }
                        }

                        // 未通过过滤器
                        if(!flag) break;

                        final StringBuilder tagsBuf = new StringBuilder();
                        for (final Map.Entry<String, String> tag
                                : Internal.getTags(tsdb, key).entrySet()) {
                            tagsBuf.append(' ').append(tag.getKey())
                                    .append('=').append(tag.getValue());
                        }

                        for (final KeyValue kv : row) {
                            formatKeyValue(buf, kv, base_time, metric, tagsBuf.toString());

                            if (buf.length() > 0) {
                                Files.append(buf, file, Charsets.UTF_8);
                                // 注意清理buf
                                buf.setLength(0);
                                if(limit.sum() <= 0) return;
                            }
                        }
                    }
                }

            }
        }
    }

    /**
     * 解析数据，不考虑appendPoint等模式，只处理时间戳为13位put模式的数据
     *
     * @param buf
     * @param kv
     * @param base_time
     * @param metric
     */
    private void formatKeyValue(final StringBuilder buf,
                                final KeyValue kv,
                                final long base_time,
                                final String metric,
                                final String tags) {


        final Collection<Internal.Cell> cells;

        cells = Internal.extractDataPoints(kv);

        for (Internal.Cell cell : cells) {
            // 判断时间
            long timeout = cell.timestamp(base_time);

            if(timeout >= dataQuery.startTime() && timeout <= dataQuery.endTime()){
                buf.append(metric)
                        .append(' ')
                        .append(cell.absoluteTimestamp(base_time))
                        .append(" ")
                        .append(cell.parseValue())
                        .append(tags)
                        .append("\n");
                limit.decrement();
            }
            if(limit.sum() <= 0) return;
        }
    }


}
